/*
 * Copyright Hundsun Technologies Inc. All Rights Reserved.
 */

package redis

import (
	"errors"
	"github.com/gomodule/redigo/redis"
	"hundsun.com/hsl/hschain/common/config"
	"hundsun.com/hsl/hschain/common/log/logging"
	"sync"
	"time"
)

var (
	errEmptyKey          = errors.New("key cannot be empty")
	errEmptyValue        = errors.New("value cannot be empty")
	errEmptyKeys         = errors.New("keys cannot be empty")
	errNotSameLength     = errors.New("keys and values must be with the same length")
	errEmptyNewValue     = errors.New("new value cannot be empty")
	errNonpositiveExpire = errors.New("expire cannot must be positive")
)

const (
	stringReplyOK    = "OK"
	compareAndSetLua = `
	if redis.call('get',KEYS[1]) == ARGV[1] then 
		return redis.call('set',KEYS[1],ARGV[2]) 
    else 
		return nil
    end       
    `
	compareAndDelLua = `
	if redis.call('get',KEYS[1]) == ARGV[1] then 
		return redis.call('del',KEYS[1]) 
    else
		return 0 
    end
   `
	setEXWhenEqualsOrNotExistsLua = `
    local oldVal = redis.call('get',KEYS[1])
    if (not oldVal) or (oldVal == ARGV[1]) then
		return redis.call('set',KEYS[1],ARGV[2],'ex',ARGV[3])
    else
		return nil
    end
    `
)

var client *Client
var once sync.Once
var log logging.Log

const (
	configKeyAddress                  = "redis.address"
	configKeyAuth                     = "redis.auth"
	configKeyDB                       = "redis.db"
	configKeyMaxIdle                  = "redis.maxIdle"
	configKeyMaxActive                = "redis.maxActive"
	configKeyIdleTimeoutMillis        = "redis.idleTimeoutMillis"
	configKeyDialConnectTimeoutMillis = "redis.dialConnectTimeoutMillis"
	configKeyDialReadTimeoutMillis    = "redis.dialReadTimeoutMillis"
	configKeyDialWriteTimeoutMillis   = "redis.dialWriteTimeoutMillis"
)

// GetClient 获取redis客户端（单例，懒加载）
func GetClient() *Client {
	once.Do(func() {
		redisConfig := &config.RedisConfig{
			Address:                  config.GetStringConfigValue(configKeyAddress, "localhost:6379"),
			Auth:                     config.GetStringConfigValue(configKeyAuth, ""),
			DB:                       config.GetStringConfigValue(configKeyDB, "0"),
			MaxIdle:                  config.GetIntConfigValue(configKeyMaxIdle, 3),
			MaxActive:                config.GetIntConfigValue(configKeyMaxActive, 10),
			IdleTimeoutMillis:        config.GetIntConfigValue(configKeyIdleTimeoutMillis, 1000),
			DialConnectTimeoutMillis: config.GetIntConfigValue(configKeyDialConnectTimeoutMillis, 3000),
			DialReadTimeoutMillis:    config.GetIntConfigValue(configKeyDialReadTimeoutMillis, 3000),
			DialWriteTimeoutMillis:   config.GetIntConfigValue(configKeyDialWriteTimeoutMillis, 3000),
		}
		log = logging.GetLogWithCommonConf(logging.ConfigKeyRedisLogLevel, logging.LogFieldRedis, logging.LogFieldRedis)

		pool := &redis.Pool{
			MaxIdle:     redisConfig.MaxIdle,
			MaxActive:   redisConfig.MaxActive,
			IdleTimeout: time.Duration(redisConfig.IdleTimeoutMillis) * time.Millisecond,
			Dial: func() (redis.Conn, error) {
				c, err := redis.Dial("tcp", redisConfig.Address,
					redis.DialConnectTimeout(time.Duration(redisConfig.DialConnectTimeoutMillis)*time.Millisecond),
					redis.DialReadTimeout(time.Duration(redisConfig.DialReadTimeoutMillis)*time.Millisecond),
					redis.DialWriteTimeout(time.Duration(redisConfig.DialWriteTimeoutMillis)*time.Millisecond))
				if err != nil {
					return nil, err
				}
				if len(redisConfig.Auth) > 0 {
					_, err = c.Do("AUTH", redisConfig.Auth)
					if err != nil {
						return nil, err
					}
				}
				_, err = c.Do("SELECT", redisConfig.DB)
				if err != nil {
					return nil, err
				}

				return c, nil
			},
			TestOnBorrow: func(c redis.Conn, t time.Time) error {
				_, err := c.Do("PING")
				return err
			},
		}

		client = &Client{
			pool: pool,
		}

		if ok, err := client.isAvailable(); !ok {
			log.Fatal(err)
			panic(err.Error())
		}

	})
	return client
}

// Client 对redigo客户端的封装
type Client struct {
	pool *redis.Pool
}

func closeConn(conn redis.Conn) {
	if err := conn.Close(); err != nil {
		log.Errorf("failed to close connection: %s", err.Error())
	}
}

// isAvailable 判断redis连接可用
func (cli *Client) isAvailable() (bool, error) {
	conn := cli.pool.Get()
	defer closeConn(conn)

	if err := conn.Err(); err != nil {
		return false, err
	}
	return true, nil
}

// Get 根据key获取value
func (cli *Client) Get(key string) ([]byte, error) {
	if len(key) == 0 {
		return nil, errEmptyKey
	}
	defer log.ElapsedTime("Get", time.Now())
	conn := cli.pool.Get()
	defer closeConn(conn)

	reply, err := conn.Do("GET", key)
	if err != nil {
		return nil, err
	}
	if reply == nil {
		return nil, nil
	}
	return redis.Bytes(reply, err)
}

// BulkGet 根据keys获取values
func (cli *Client) BulkGet(keys []string) ([][]byte, error) {
	if len(keys) == 0 {
		return nil, errEmptyKeys
	}
	defer log.ElapsedTime("BulkGet", time.Now())
	conn := cli.pool.Get()
	defer closeConn(conn)
	return redis.ByteSlices(conn.Do("MGET", redis.Args{}.AddFlat(keys)...))
}

// Set 向缓存中放入一个KV对。bool类型返回值表示是否成功存入。
func (cli *Client) Set(key string, value []byte) (bool, error) {
	if len(key) == 0 {
		return false, errEmptyKey
	}
	if len(value) == 0 {
		return false, errEmptyValue
	}

	defer log.ElapsedTime("Set", time.Now())

	return setAndReturnBool(cli, key, value)
}

func setAndReturnBool(cli *Client, args ...interface{}) (bool, error) {
	conn := cli.pool.Get()
	defer closeConn(conn)
	reply, err := conn.Do("SET", args...)
	if err != nil || reply == nil {
		return false, err
	}

	replyStr, err := redis.String(reply, err)
	return replyStr == stringReplyOK, nil
}

// Delete 删除指定的key。返回值是删除的条数
func (cli *Client) Delete(key string) (int, error) {
	defer log.ElapsedTime("Delete", time.Now())
	conn := cli.pool.Get()
	defer closeConn(conn)
	return redis.Int(conn.Do("DEL", key))
}

// BulkDelete 批量删除指定的keys；返回删除的条数
func (cli *Client) BulkDelete(keys []string) (int, error) {
	defer log.ElapsedTime("BulkDelete", time.Now())
	conn := cli.pool.Get()
	defer closeConn(conn)
	return redis.Int(conn.Do("DEL", redis.Args{}.AddFlat(keys)...))
}

// SetEX 赋值并设置key过期时间
func (cli *Client) SetEX(key string, value []byte, expireSeconds int) (bool, error) {
	if len(key) == 0 {
		return false, errEmptyKey
	}
	if len(value) == 0 {
		return false, errEmptyValue
	}

	defer log.ElapsedTime("SetEX", time.Now())

	return setAndReturnBool(cli, key, value, "EX", expireSeconds)
}

// BulkSetEX 批量赋值并设置key过期时间
func (cli *Client) BulkSetEX(keys []string, values [][]byte, expireSeconds int) error {
	if len(keys) == 0 {
		return nil
	}
	if len(keys) != len(values) {
		return errNotSameLength
	}

	defer log.ElapsedTime("BulkSetEX", time.Now())
	conn := cli.pool.Get()
	defer closeConn(conn)

	// pipeline: send + flush
	for i, key := range keys {
		_ = conn.Send("SET", key, values[i], "EX", expireSeconds)
	}
	if err := conn.Flush(); err != nil {
		return err
	}
	if _, err := conn.Do(""); err != nil {
		// 忽略返回值
		return err
	}
	return nil
}

// BulkSet 批量赋值
func (cli *Client) BulkSet(keys []string, values [][]byte) error {
	if len(keys) == 0 {
		return nil
	}
	if len(keys) != len(values) {
		return errNotSameLength
	}

	defer log.ElapsedTime("BulkSet", time.Now())
	conn := cli.pool.Get()
	defer closeConn(conn)

	args := redis.Args{}
	for i, key := range keys {
		args = args.Add(key, values[i])
	}
	_, err := redis.String(conn.Do("MSET", args...))
	return err
}

// SetNX 不存在时才赋值
func (cli *Client) SetNX(key string, value []byte) (bool, error) {
	if len(key) == 0 {
		return false, errEmptyKey
	}
	if len(value) == 0 {
		return false, errEmptyValue
	}
	defer log.ElapsedTime("SetNX", time.Now())
	conn := cli.pool.Get()
	defer closeConn(conn)
	return redis.Bool(conn.Do("SETNX", key, value))
}

// SetNXEX 不存在时才赋值并设置key过期时间
func (cli *Client) SetNXEX(key string, value []byte, expireSeconds int) (bool, error) {
	if len(key) == 0 {
		return false, errEmptyKey
	}
	if len(value) == 0 {
		return false, errEmptyValue
	}
	if expireSeconds < 0 {
		return false, errNonpositiveExpire
	}
	return setAndReturnBool(cli, key, value, "EX", expireSeconds, "NX")
}

// Expire 对已有key设置过时间期（负数或0表示不过期）
func (cli *Client) Expire(keys []string, expireSeconds int) ([]int64, error) {
	if len(keys) == 0 {
		return nil, errEmptyKeys
	}
	defer log.ElapsedTime("Expire", time.Now())
	conn := cli.pool.Get()
	defer closeConn(conn)

	if err := conn.Send("MULTI"); err != nil {
		return nil, err
	}
	// 为保持返回数组长度一致，无需判断key为nil
	if expireSeconds < 1 {
		for _, key := range keys {
			_ = conn.Send("PERSIST", key)
		}
	} else {
		for _, key := range keys {
			_ = conn.Send("EXPIRE", key, expireSeconds)
		}
	}

	return redis.Int64s(conn.Do("EXEC"))
}

// CompareAndSet 如果key对应的值为value，则将值设置为newValue
func (cli *Client) CompareAndSet(key string, value []byte, newValue []byte) (bool, error) {
	if len(key) == 0 {
		return false, errEmptyKey
	}
	if len(value) == 0 {
		return false, errEmptyValue
	}
	if len(newValue) == 0 {
		return false, errEmptyNewValue
	}
	defer log.ElapsedTime("CompareAndSet", time.Now())
	conn := cli.pool.Get()
	defer closeConn(conn)
	script := redis.NewScript(1, compareAndSetLua)
	if reply, err := script.Do(conn, key, value, newValue); err != nil {
		return false, err
	} else if reply == nil {
		return false, nil
	} else {
		return reply == stringReplyOK, nil
	}
}

// CompareAndDelete 如果key对应的值为value，则删除key。
func (cli *Client) CompareAndDelete(key string, value []byte) (bool, error) {
	if len(key) == 0 {
		return false, errEmptyKey
	}
	if len(value) == 0 {
		return false, errEmptyValue
	}

	defer log.ElapsedTime("CompareAndDelete", time.Now())
	conn := cli.pool.Get()
	defer closeConn(conn)
	script := redis.NewScript(1, compareAndDelLua)
	return redis.Bool(script.Do(conn, key, value))
}

// SetEXWhenEqualsOrNotExists 如果key对应的值为value或者空，则将值设置为newValue并设置过期时间
func (cli *Client) SetEXWhenEqualsOrNotExists(key string, value []byte, newValue []byte, expireSeconds int) (bool, error) {
	if len(key) == 0 {
		return false, errEmptyKey
	}
	if len(value) == 0 {
		return false, errEmptyValue
	}
	if len(newValue) == 0 {
		return false, errEmptyNewValue
	}
	if expireSeconds < 0 {
		return false, errNonpositiveExpire
	}
	defer log.ElapsedTime("SetEXWhenEqualsOrNotExists", time.Now())
	conn := cli.pool.Get()
	defer closeConn(conn)
	script := redis.NewScript(1, setEXWhenEqualsOrNotExistsLua)
	if reply, err := script.Do(conn, key, value, newValue, expireSeconds); err != nil {
		return false, err
	} else if reply == nil {
		return false, nil
	} else {
		return reply == stringReplyOK, nil
	}
}

// Shutdown 关闭redis客户端
func (cli *Client) Shutdown() {
	err := cli.pool.Close()
	if err != nil {
		log.Errorf("failed to close redis client: %s", err.Error())
	}
}

// Sadd 集合-新增，返回成功添加个数
func (cli *Client) Sadd(setName string, keys []string) (int, error) {
	defer log.ElapsedTime("Sadd", time.Now())
	conn := cli.pool.Get()
	defer closeConn(conn)

	return redis.Int(conn.Do("SADD", redis.Args{}.Add(setName).AddFlat(keys)...))
}

// Spop 集合-返回元素
func (cli *Client) Spop(setName string, count int) ([]string, error) {
	defer log.ElapsedTime("Spop", time.Now())
	conn := cli.pool.Get()
	defer closeConn(conn)

	return redis.Strings(conn.Do("SPOP", setName, count))
}

// Smove 集合-移动元素
func (cli *Client) Smove(fromSetName string, toSetName string, keys []string) error {
	defer log.ElapsedTime("Smove", time.Now())
	conn := cli.pool.Get()
	defer closeConn(conn)

	// pipeline
	for _, key := range keys {
		_ = conn.Send("SMOVE", fromSetName, toSetName, key)
	}
	if err := conn.Flush(); err != nil {
		return err
	}
	if _, err := conn.Do(""); err != nil {
		// 忽略返回值
		return err
	}
	return nil
}

// Srem 集合-删除，返回成功删除个数
func (cli *Client) Srem(setName string, keys []string) (int, error) {
	defer log.ElapsedTime("Srem", time.Now())
	conn := cli.pool.Get()
	defer closeConn(conn)

	return redis.Int(conn.Do("SREM", redis.Args{}.Add(setName).AddFlat(keys)...))
}
